package utils

import (
	"errors"
	"fmt"
	"os"
	"reflect"
	"sync/atomic"
	"time"
)

/*
**

	如果使用chan 做纤程进入计数,
	  不能中途改变 允许进入纤程大小
*/
type multi_group_task_channel struct {
	state                 int32 // 0 可用, 1不可用
	task_current_busy_cnt int32
	task_current_work_cnt int32  // 当前正在工作的纤程数
	work_max_cnt          int32  // 最大允许的工作纤程数量
	push_cnt              uint32 // 压入次数
	queue_size            int32  // 当前任务数量
	max_queue_size        int32  // 最大的运行队列数量
	push_err_n            int32  // 压入失败数量
	id                    interface{}
	dataQueue             *SyncQueue // 任务队列

	last_activity     time.Time // 最后活动时间
	last_push_T       time.Time
	last_pop_T        time.Time
	last_exec_start_T time.Time

	last_goroute_id uint64

	// 最后压入的任务
	last_push_task *MultiGroupTaskRec

	// 最后压入失败的任务
	last_push_fail_task *MultiGroupTaskRec

	entryInfo *SyncCycleList // 调试时使用 跟踪进入的信息

	tagstr string
}

func (this *multi_group_task_channel) reset() {
	this.state = 0
	this.work_max_cnt = 1 // 默认为1
	this.task_current_work_cnt = 0
	this.push_cnt = 0
	this.queue_size = 0
	this.push_err_n = 0
}

func (this *multi_group_task_channel) Push(rec *MultiGroupTaskRec) error {
	if this.state != 0 {
		atomic.AddInt32(&this.push_err_n, 1)
		return errors.New(fmt.Sprintf("队列[%v]被标记为不可用,请重新进行压入！", this.id))
	}

	r := atomic.AddInt32(&this.queue_size, 1)
	if r > this.max_queue_size { // 失败
		atomic.AddInt32(&this.queue_size, -1)
		atomic.AddInt32(&this.push_err_n, 1)
		if DEBUG_MODE == 1 {
			if len(rec.CallStack) == 0 {
				rec.CallStack = GetCallStack(2)
			}
		}
		this.last_push_fail_task = rec
		return errors.New(fmt.Sprintf("队列[%v]超过堆积数量[%d]！", this.id, this.max_queue_size))
	}
	atomic.AddUint32(&this.push_cnt, 1)
	this.dataQueue.Push(rec)
	this.last_push_T = time.Now()
	this.last_push_task = rec
	return nil
}

func (this *multi_group_task_channel) Pop() *MultiGroupTaskRec {
	ok, rval := this.dataQueue.Pop()
	if ok {
		atomic.AddInt32(&this.queue_size, -1)
		this.last_pop_T = time.Now()
		return rval.(*MultiGroupTaskRec)
	} else {
		return nil
	}
}

func (this *multi_group_task_channel) String() string {
	var jb JSONBuilder
	jb.ObjectBegin()
	strid := LeftStr(fmt.Sprintf("%v", this.id), 32)
	if len(strid) > 0 {
		jb.Add("id", strid).Spliter()
	}
	jb.AddValues("current-work-n", this.task_current_work_cnt, "max-allow-work-n", this.work_max_cnt, "remain", this.queue_size, "push-err-n", this.push_err_n)
	jb.ObjectEnd()
	return jb.String()
}

func (this *multi_group_task_channel) GetBusyDuration() time.Duration {
	busycnt := atomic.LoadInt32(&this.task_current_busy_cnt)
	if busycnt == 0 {
		return 0
	}
	return time.Now().Sub(this.last_exec_start_T)
}

func (this *multi_group_task_channel) IDText() string {
	if this.id == nil {
		return "NIL"
	} else {
		switch vv := this.id.(type) {
		case string:
			return LeftStr(vv, 64)
		case int, int8, int32, int64, uint, uint32, uint8:
			return fmt.Sprintf("%d", vv)
		case float64, float32:
			return fmt.Sprintf("%v", vv)
		default:
			return LeftStr(fmt.Sprintf("%s:%v", reflect.TypeOf(this.id).String(), this.id), 64)
		}
	}
}

func (this *multi_group_task_channel) IsAbnormal() bool {
	return this.GetBusyDuration().Seconds() > 10
}

func (this *multi_group_task_channel) StatusString() string {
	var sb BytesBuilder
	sb.Appendf("id: %s\n", this.IDText())
	if this.last_goroute_id != 0 {
		sb.Appendf("lastgoid:%d\n", this.last_goroute_id)
	}
	sb.Appendf("last_exec_start_t: %s (%d ms)\n", DateTimeString2(this.last_exec_start_T), this.GetBusyDuration().Milliseconds())
	sb.Appendf("last_activity:%s\n", DateTimeString2(this.last_activity))
	sb.Appendf("work: [%d], %d/%d(max)\n", this.task_current_busy_cnt, this.task_current_work_cnt, this.work_max_cnt)
	sb.Appendf("task: %d (remain), %d(push fail)\n", this.queue_size, this.push_err_n)
	sb.Appendf("tagstr:%s\n", this.tagstr)
	t1 := this.last_push_fail_task
	if t1 != nil {
		sb.Appendf("last_push_fail_callstack:%s\n", t1.CallStack)
	}

	t2 := this.last_push_task
	if t2 != nil && len(t2.CallStack) > 0 {
		sb.Appendf("last_push_callstack:%s\n", t2.CallStack)
	}

	return sb.String()
}

func (this *multi_group_task_channel) needWorker() bool {
	if this.task_current_work_cnt >= this.work_max_cnt {
		return false
	}

	if this.queue_size == 0 {
		return false
	}

	return true
}

/*
**
  - 开始工作
  - 2. 开启工作后, Push可以正常加入
*/
func (this *multi_group_task_channel) tryBeginWork() bool {
	if atomic.LoadInt32(&this.queue_size) == 0 { // 不需要工作
		return false
	}

	if this.work_max_cnt == 0 { // 没有限制工作纤程的进入
		r := atomic.AddInt32(&this.task_current_work_cnt, 1)
		if DEBUG_MODE == 1 {
			this.entryInfo.Push(fmt.Sprintf("+[%d]tryBeginWork :%d (%d)", GetCurrentGoRoutineID(), r, this.task_current_work_cnt))
		}
		return true
	}

	if atomic.LoadInt32(&this.task_current_work_cnt) >= this.work_max_cnt { // 工作纤程已经达到了最大值
		return false
	}

	r := atomic.AddInt32(&this.task_current_work_cnt, 1)
	if DEBUG_MODE == 1 {
		this.entryInfo.Push(fmt.Sprintf("+[%d]tryBeginWork :%d (%d)", GetCurrentGoRoutineID(), r, this.task_current_work_cnt))
	}
	if r <= this.work_max_cnt {
		return true
	} else {
		atomic.AddInt32(&this.task_current_work_cnt, -1)
		if DEBUG_MODE == 1 {
			this.entryInfo.Push(fmt.Sprintf("-[%d]tryBeginWork fail :%d (%d)", GetCurrentGoRoutineID(), r, this.task_current_work_cnt))
		}
		return false
	}
}

/**
 * 关闭工作
 */
func (this *multi_group_task_channel) endWork() bool {
	x1 := this.task_current_work_cnt
	r := atomic.AddInt32(&this.task_current_work_cnt, -1)
	if DEBUG_MODE == 1 {
		this.entryInfo.Push(fmt.Sprintf("-[%d]endWork : %d -> %d (%d)", GetCurrentGoRoutineID(), x1, r, this.task_current_work_cnt))
	}
	if r == 0 {
		this.last_activity = time.Now()
	}

	if r < 0 {
		if DEBUG_MODE == 1 {
			var debuginfo string
			this.entryInfo.Range(func(args ...interface{}) bool {
				debuginfo += args[0].(string) + "\r\n"
				return true
			})
			fmt.Fprintf(os.Stderr, "[BUG][%v]endWork cnt err:%d, (%d), debuginfo \n %s\r\n", this.id, r, this.task_current_work_cnt, debuginfo)
		} else {
			fmt.Fprintf(os.Stderr, "[BUG][%v]endWork cnt err:%d, (%d)\r\n", this.id, r, this.task_current_work_cnt)
		}

	}

	return true
}

/***
 * true  表示可以进行清理
 * false 表示不可以进行清理,或者已经被清理
 */
func (this *multi_group_task_channel) tryMarkCleanup(secs float64) bool {
	if this.queue_size > 0 {
		return false
	}

	if this.task_current_work_cnt >= 1 { // 有工作纤程进入
		return false
	}

	if this.last_activity.IsZero() {
		return false
	}

	if time.Now().Sub(this.last_activity).Seconds() > secs {
		if atomic.CompareAndSwapInt32(&this.state, 0, 1) {
			if this.queue_size > 0 { // 有队列进入
				atomic.StoreInt32(&this.state, 0) // 标记为可用, 不可清理
				return false
			}
			return true
		} else {
			return false
		}
	} else {
		return false
	}
}
